package com.oracle.calllog.producer;

import com.oracle.common.PropertiesUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.FileInputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.util.Date;

/******************************************
 * @ClassName CallLogsHDFSProducer
 * @Description TODO
 * @Author WolfSoul
 * @Date 2022/6/8 22:46
 * @Version 1.0
 *****************************************/
public class CallLogs2DfsProducer extends CallLogsStandardSimulationProducer {

    /**
     * 用于生成模拟异常数据(即垃圾数据)的方法<BR>
     * 异常通信数据种类描述:<BR>
     * ERROR-0 = 基站通信线程同步异常产生的数据都为NULL信息的通话日志<BR>
     * ERROR-1 = 被叫人号码非常规号码(未知号码)的通话日志<BR>
     * ERROR-2 = 没有接通状态与通话时长信息的通话日志<BR>
     * ERROR-3 = 只有主叫人信息的通话日志<BR>
     *
     * @return String 一条通话日志的垃圾数据
     */
    protected String generateRubbishDatas(String callLog) {
        //生成随机垃圾数据种类
        int rubbishType = random.nextInt(4);

        String[] callLogFields = callLog.split(" ");
        switch (rubbishType) {
            case 0:
                callLog = callLogFields[0] + " NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL";
                break;
            case 1:
                callLog = callLogFields[0] + " " + callLogFields[1] + " " + callLogFields[2] + " "
                        + callLogFields[3] + " " + callLogFields[4] + " " + callLogFields[5] + " "
                        + "* * * " + this.generateRandomLengthNumber() + " * "
                        + callLogFields[11] + " 0000 LOSS_CALL";
                break;
            case 2:
                callLog = callLogFields[0] + " " + callLogFields[1] + " " + callLogFields[2] + " "
                        + callLogFields[3] + " " + callLogFields[4] + " " + callLogFields[5] + " "
                        + callLogFields[6] + " " + callLogFields[7] + " " + callLogFields[8] + " "
                        + callLogFields[9] + " " + callLogFields[10] + " " + callLogFields[11];
                break;
            case 3:
                callLog = callLogFields[0] + " " + callLogFields[1] + " " + callLogFields[2] + " "
                        + callLogFields[3] + " " + callLogFields[4] + " " + callLogFields[5] + " "
                        + "     " + callLogFields[11] + " 0000 LOSS_CALL";
        }
        return callLog;

    }

    /**
     * 用于生成一个随机长度(最多22位)的数字字符串,用于表示错误的呼叫号码
     *
     * @return
     */
    protected String generateRandomLengthNumber() {
        StringBuilder builder = new StringBuilder();
        int length = random.nextInt(22) + 1;
        //用于验证标准手机号码的正则表达式,号段规则如下(截止2021年春节)
        //    	13 0123456789
        //    	14 145678
        //    	15 012356789
        //    	16 567
        //    	17 0145678
        //    	18 0123456789
        //    	19 189
        String mobileRegExp = "1((3[0-9])|(4[14-8])|(5[0-35-9])|(6[5-7])|(7[014-8])|(8[0-9])|(9[189]))\\d{8}";
        do {
            for (int i = 0; i < length; i++) {
                builder.append(random.nextInt(10));
            }
            if (length != 11) break;//减少正则验证压力,长度不是11则非手机号码
        } while (builder.toString().matches(mobileRegExp));
        return builder.toString();
    }

    /**
     * 将数据写入到文件中
     *
     * @param strScale       随机联系人的体量
     * @param strStartDate   字符串格式起始日期,格式为:yyyy-MM-dd
     * @param strEndDate     字符串格式起始日期,格式为:yyyy-MM-dd
     */
    public void writeLog(String strScale, String strStartDate, String strEndDate) {
        Configuration configuration = new Configuration();
        // 这个解决hdfs问题
        configuration.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
        // 这个解决本地file问题
        configuration.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());


        String hdfsAddr = PropertiesUtils.getProperty("hdfs.address");
        String hdfsOutputPath = PropertiesUtils.getProperty("hdfs.output.path");
        String hdfsUser = PropertiesUtils.getProperty("hdfs.user");
        String hdfsFilesCount = PropertiesUtils.getProperty("hdfs.files.count");
        String hdfsFileSize = PropertiesUtils.getProperty("hdfs.file.size");

        int iFilesCount = Integer.valueOf(hdfsFilesCount);
        int iFileSize = Integer.valueOf(hdfsFileSize);
        int filesCount = 0;

        long currentTime = System.currentTimeMillis();
        String fileName = "calllog-"+currentTime+".log";
        String tmpFileName = "calllog-"+currentTime+".log.tmp";

        if (!"/".equals(hdfsOutputPath) && !hdfsOutputPath.endsWith("/")) {
            hdfsOutputPath += "/";
        }
        String tmpOutputPath = "";

        if(!"/".equals(hdfsOutputPath)) {
            String parentPath = hdfsOutputPath.substring(0, hdfsOutputPath.length()-1);
            tmpOutputPath = parentPath.substring(0, parentPath.lastIndexOf("/")) + "/logs-tmp";
        }else{
            hdfsOutputPath = "/calllog-datas";
            tmpOutputPath = "/calllog-tmp";
        }

        System.out.println("HDFS==>" + hdfsAddr);

        String copyLogsPath = hdfsAddr + hdfsOutputPath;
        String writeLogsPath = hdfsAddr + tmpOutputPath;
        Path writePath = new Path(copyLogsPath);
        Path tmpPath = new Path(writeLogsPath);
        Path finalFile = new Path(copyLogsPath +"/"+ fileName);
        Path tmpFile = new Path(writeLogsPath +"/"+ tmpFileName);

        System.out.println("Datas OutPath==>" + copyLogsPath);

        int times = 1;

        FileSystem fs = null;
        FileInputStream inStream = null;
        FSDataOutputStream outStream = null;

        try {
            fs = FileSystem.get(new URI(hdfsAddr), configuration, hdfsUser);

            if(!fs.exists(writePath)){
                fs.mkdirs(writePath);
            }
            if(!fs.exists(tmpPath)){
                fs.mkdirs(tmpPath);
            }

            outStream = fs.create(tmpFile,true);

            while (true) {
                //每1500毫秒随机产生0-100条通话记录
                Thread.sleep(500);

                int count = random.nextInt(1000);
                for (int i = 0; i < count; i++) {
                    String log = this.product(strScale, strStartDate, strEndDate);
                    if (random.nextInt(10) == 1) {
                        log = this.generateRubbishDatas(log);
                    }

                    System.out.println(log);

                    outStream.write((log + "\r\n").getBytes("UTF-8"));

                    if(outStream.size()>1024*1024*times){
                        outStream.hsync();
                        times++;
                    }

                }

                //控制写出文件大小
                if(outStream.size()>=1048576L*iFileSize){
                    outStream.hsync();
                    outStream.close();
                    fs.rename(tmpFile, finalFile);

                    times = 1;
                    filesCount++;
                    if(filesCount == iFilesCount){
                        break;
                    }

                    currentTime = System.currentTimeMillis();
                    fileName = "calllog-"+currentTime+".log";
                    tmpFileName = "calllog-"+currentTime+".log.tmp";
                    finalFile = new Path(copyLogsPath +"/"+ fileName);
                    tmpFile = new Path(writeLogsPath +"/"+ tmpFileName);

                    outStream = fs.create(tmpFile,true);

                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        } finally {
            if (outStream != null) {
                try {
                    outStream.hflush();
                    outStream.close();
                    fs.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }


    }

    /**
     * 仿真系统运行入口主方法
     *
     * @param args 数据存储路径[ 通话数据生成范围的起始日期 通话数据生成范围的截止日期]
     * @throws InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
        String scale = "any";
        String startDate = "2021-01-01";
        //默认终止时间点为当前时间的上一天
        long endTime = System.currentTimeMillis() - 24 * 60 * 60 * 1000;
        String endDate = new SimpleDateFormat("yyyy-MM-dd").format(new Date(endTime));

        // 传递1个参数时，数据规模根据参数决定(any|all)，否则默认为any
        if (args!=null && args.length >= 1)
            scale = args[0];
        // 传递2个参数时，随机通话时间数据的起始时间点使用传递的参数，终止时间点使用默认值
        if (args!=null && args.length >= 2)
            startDate = args[1];
        //传递3个以上参数时，随机通话时间数据的起始和终止时间点都使用传递的参数
        if (args!=null && args.length >= 3)
            endDate = args[2];

        CallLogs2DfsProducer productLog = new CallLogs2DfsProducer();
        productLog.init(scale);
        productLog.writeLog(scale, startDate, endDate);
    }
}
